package nr.nodehost;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import lombok.Getter;
import org.springframework.stereotype.Component;

import net.sf.json.JSONObject;


@Component
public class Dispatcher {

	@Getter
	private HashMap<String,ComponentExecutor> componentExecutorMap=new HashMap<String,ComponentExecutor>();
	private HashMap<Integer,SchemaExecutor> schemaExecutorMap=new HashMap<>();
	public void init() {
		System.out.println("dispatcher inited");
	}
	public void dataArrive(DataFrame frame) {
		//System.out.println(frame.toString());
		int schemaId=frame.getSchemaId();
		SchemaExecutor schemaExecutor=this.getSchemaExecutor(schemaId);
		if(schemaExecutor!=null) {

			ComponentExecutor componentExecutor=componentExecutorMap.get(frame.getTargetComponentId());
			if(componentExecutor!=null)
				componentExecutor.dataArrived(frame);
			else
				System.out.println("component "+frame.getTargetComponentId()+" is not ready for receive data");
		}
		else {
			System.out.println("schema "+schemaId+" isn't running ");
			frame=null;
			System.gc();
		}

	}

	public void registerComponentExecutor(ComponentExecutor componentExecutor) {
		componentExecutorMap.put(componentExecutor.getComponentId(), componentExecutor);
	}
	
	public void registerSchemaExecutor(SchemaExecutor schemaExecutor) {
		schemaExecutorMap.put(schemaExecutor.getSchemaId(),schemaExecutor);
	}

	public JSONObject stopSchemaExecutor(int schemaId) {
		JSONObject result=new JSONObject();
		SchemaExecutor executor=schemaExecutorMap.get(schemaId);
		if (executor!=null){
//			if(executor.getStatus()!=SchemaExecutor.IDLE) {

			System.out.println("############executor status:"+executor.getStatus());
			if(executor.terminate_docker()) {
				result.put("code", 0);
				result.put("info","stop schema "+schemaId+" success.");
			}
			else {
				result.put("code", 1);
				result.put("info","stop schema "+schemaId+" fail.");
			}

//			else {
//				System.out.println("schema "+schemaId+" isn't running ");
//				result.put("code", 2);
//				result.put("info","schema "+schemaId+" isn't running ");
//			}
			schemaExecutorMap.remove(schemaId);
		}

		return result;
	}
	
//	public JSONObject stopSchemaExecutor(int schemaId) {
//		JSONObject result=new JSONObject();
//		SchemaExecutor executor=schemaExecutorMap.get(schemaId);
//		if (executor!=null){
//			if(executor.getStatus()!=SchemaExecutor.IDLE) {
//				System.out.println(executor.getStatus());
//				if(executor.terminate()) {
//					result.put("code", 0);
//					result.put("info","stop schema "+schemaId+" success.");
//				}
//				else {
//					result.put("code", 1);
//					result.put("info","stop schema "+schemaId+" fail.");
//				}
//			}
//		else {
//				System.out.println("schema "+schemaId+" isn't running ");
//				result.put("code", 2);
//				result.put("info","schema "+schemaId+" isn't running ");
//			}
//			schemaExecutorMap.remove(schemaId);
//		}
//
//		return result;
//	}
	
	public JSONObject pauseSchemaExecutor(int schemaId) {
		JSONObject result=new JSONObject();
		SchemaExecutor executor=schemaExecutorMap.get(schemaId);
		if(executor!=null&&executor.getStatus()==SchemaExecutor.RUNNING) {
			if(executor.pause())
				result.put("code", 0);
			else {
				result.put("code", 1);
				result.put("info","pause schema "+schemaId+" fail.");
			}
		}
		else {
			System.out.println("schema "+schemaId+" isn't running ");
			result.put("code", 2);
			result.put("info","schema "+schemaId+" isn't running ");
		}
		return result;
	}
	
	public JSONObject resumeSchemaExecutor(int schemaId) {
		JSONObject result=new JSONObject();
		SchemaExecutor executor=schemaExecutorMap.get(schemaId);
		if(executor!=null&&executor.getStatus()==SchemaExecutor.PAUSED) {
			if(executor.resume())
				result.put("code", 0);
			else {
				result.put("code", 1);
				result.put("info","resume schema "+schemaId+" fail.");
			}
		}
		else {
			System.out.println("schema "+schemaId+" isn't paused ");
			result.put("code", 2);
			result.put("info","schema "+schemaId+" isn't paused ");
		}
		return result;
	}
	
	public SchemaExecutor getSchemaExecutor(int schemaId) {
		return schemaExecutorMap.get(schemaId);
	}
	

	public void processMessage(JSONObject message) {
		try{
			String messageType=message.getString("messageType");

			JSONObject messageContent=message.getJSONObject("messageContent");
			if(messageType.equals("dataMessage")) {
				processDataMessage(messageContent);
			}
		}
		catch(Exception e) {
			e.printStackTrace();
		}
	}

	private void processDataMessage(JSONObject message) {
		DataFrame outFrame=new DataFrame(
				message.getInt("frameNo"),//targetComponent.getString("id").equals(param.getString("uuid"))?nextFrameNo+1:nextFrameNo,
				message.getInt("schemaId"),
				message.getString("sourceComponentId"),
				message.getString("sourceComponentPort"),
				message.getString("targetComponentId"),
				message.getString("targetComponentPort"),
				message.getString("dataFilename"));
		dataArrive(outFrame);
	}
}
